package com.spring.reactive;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.function.Supplier;
import java.util.stream.Stream;

public class OperateApiTest {

    /**
     * 结果
     * [1, 2]
     * [3, 4]
     * [5, 6]
     */
    public static void test_01(){
        Flux.just(1,2,3,4,5,6).buffer(2).subscribe(System.out::println);
    }
    /**
     * [1, 2, 3, 4]
     * [3, 4, 5, 6]
     * [5, 6]
     */
    public static void test_02(){
        Flux.just(1,2,3,4,5,6).buffer(4, 2).subscribe(System.out::println);
    }

    /**
     *  有疑问 没搞懂在做什么
     */
    public static void test_03(){

        Collection<Integer> list = new ArrayList<>();

        list.add(9);
        list.add(9);
        list.add(9);
        list.add(9);
        list.add(9);
        list.add(9);
        list.add(9);

        Supplier<Collection<Integer>> supplier = ()-> list;

        Flux.just(1,2,3,4,5,6).buffer(20,supplier).subscribe(System.out::println);
    }

    /**
     * 1
     * 2
     * --------------------
     * 3
     * 4
     * --------------------
     * 5
     * 6
     * --------------------
     */
    public static void test_04(){

        Flux.just(1,2,3,4,5,6).window(2).toIterable().forEach(f->{
            f.subscribe(System.out::println);
            System.out.println("--------------------");
        });
    }

    public static void test_05(){
        Flux.just(1,2,3,4,5,6).window(2).flatMap(w-> w).subscribe(System.out::println);
    }

    /**
     *  flatmap()
     */
    public static void test_06(){
        Flux.just(1,2,3,4,5,6).map(i->i*i).subscribe(System.out::println);
    }

    /**
     *  Java lambda中 flatMap
     */
    public static void test_07(){
        String[] words = new String[]{"hello", "world"};

        Arrays.stream(words)
                // 将字符串切割成一个Stream<String[]>
                .map(s->s.split(""))
                // 将Stream<String[]> 转成 Stream<String> 并组合
                .flatMap(Stream::of)
                .forEach(System.out::println);
    }
    /**
     * 2
     * 4
     * 6
     */
    public static void test_08(){
        Flux.just(1,2,3,4,5,6).filter(i->i%2==0).subscribe(System.out::println);
    }

    /**
     *  没有任何打印
     */
    public static void test_09(){
        Flux.just(1,2,3,4,5,6).filterWhen(i->{
            System.out.println(i);
            return s->{
                s.onNext(true);
            };
        }).subscribe(System.out::println);
    }

    public static void test_10(){
        // 取第一个元素
        System.out.println(Flux.just(1, 2, 3, 4, 5, 6).blockFirst());
        // 取最后一个元素
        System.out.println(Flux.just(1, 2, 3, 4, 5, 6).blockLast());
        // 跳过两个元素
        System.out.println(Flux.just(1, 2, 3, 4, 5, 6).skip(2).blockFirst());
        // 取出序列中的两个
        Flux.just(1, 2, 3, 4, 5, 6).take(2).subscribe(System.out::println);
    }

    public static void test_11(){

        Flux.just(1, 2, 3, 4, 5, 6)
                // 返回 Flux<Void>
                .then()
                .subscribe(System.out::println);
    }

    /**
     *  处理完一个元素的 when之后 再处理下一个元素
     */
    public static void test_12(){

        Flux.just(1, 2, 3, 4, 5, 6)
                .flatMap(i->{
                    System.out.println(i);
                    return Mono.when(s->{
                        System.out.println("all done");
                        s.onComplete();
                    },
                    s->{
                        System.out.println("all done-2");
                        s.onComplete();
                    });
                }).subscribe();
    }

    /**
     *  两个序列交替合并
     */
    public static void test_13(){

        Flux.merge(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(2),
                Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(2))
                .subscribe(System.out::println);

    }

    /**
     * 按顺序合并序列
     */
    public static void test_14(){
        Flux.mergeSequential(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(3),
                Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(2))
                .subscribe(System.out::println);
    }

    /**
     * @Description: 两个序列是一对一匹配合并的， 匹配不上的丢弃
     * @param @param
     * @return @return
     * @author TongRui乀
     * @throws
     * @date 2021/3/31 3:38 下午
     */
    public static void test_15(){
        Flux.zip(Flux.just(1, 2, 5), Flux.just(3, 4)).subscribe(System.out::println);
    }

    public static void test_16(){

        Flux.just(1,2,3).mergeWith(Flux.just(4,5,6)).subscribe(System.out::println);

    }

    public static void test_17(){
        Flux.just(1,3,5).zipWith(Flux.just(2,4,6), (s1,s2)->String.format("%s -- %s", s1,s2)).subscribe(System.out::println);
    }

    /**
     *  异常处理  onErrorReturn
     */
    public static void test_18(){

        Flux.range(1, 6)
                .map(i -> 10/(i-3))
                .onErrorReturn(5)   // 出错后给一个默认值 然后序列就终端了 元素序列只剩出错前的一段
                .map(i -> i*i)
                .subscribe(System.out::println, System.err::println);
    }

    /**
     *  subscribe  第二个参数可以打印错误的信息
     */
    public static void test_19(){

        Flux.range(1, 6)
                .map(i -> 10/(i-3))
                .map(i -> i*i)
                .subscribe(System.out::println, System.err::println);
    }

    /**
     *  异常处理  onErrorResume
     *   处理异常信息，并返回值一个替代值
     */
    public static void test_20(){

        Flux.range(1, 6)
                .map(i -> 10/(i-3))
                .onErrorResume(e->{
                    e.printStackTrace();
                    return Mono.just(10);
                })
                .map(i -> i*i)
                .subscribe(System.out::println, System.err::println);
    }

    /**
     *  异常处理 onErrorMap  抛出异常
     */
    public static void test_21(){

        Flux.range(1, 6)
                .map(i -> 10/(i-3))
                .onErrorMap(e-> {throw new RuntimeException("111");})
                .map(i -> i*i)
                .subscribe(System.out::println, System.err::println);

    }

    /**
     *  异常处理 doOnError
     */
    public static void test_22(){

        Flux.range(1, 6)
                .map(i -> 10/(i-3))
                .doOnError(e->{
                    System.out.println(e.getLocalizedMessage());
                })
                .onErrorMap(e-> {throw new RuntimeException("111");})
                .map(i -> i*i)
                .subscribe(System.out::println, System.err::println);

    }

    /**
     *  如果元素序列为空的话 defautlEmpty  返回一个默认值
     */
    public static void test_23(){

        Flux.just().defaultIfEmpty(0).subscribe(System.out::println);

    }


    /**
     *  take 从元素序列中取元素 直到满足条件
     */
    public static void test_24(){

        Flux.range(1, 100)
                .takeUntil(i->i == 10)
                .subscribe(System.out::println);

    }

    /**
     *  结果同上， 当满足条件时 才会从序列中取元素
     */
    public static void test_25(){

        Flux.range(1, 100)
                .takeWhile(i->i <= 10)
                .subscribe(System.out::println);

    }

    /**
     *  满足条件时才开始从序列中提取元素
     */
    public static void test_26(){

        Flux.range(1, 100)
                .skipUntil(i->i > 10)
                .subscribe(System.out::println);
    }

    /**
     *  跳过满足条件的元素
     */
    public static void test_27(){

        Flux.range(1, 100)
                .skipWhile(i->i <= 10)
                .subscribe(System.out::println);

    }

    /**
     *  元素序列中有任意一个满足any条件的 都将返回一个ture 
     */
    public static void test_28(){
        Flux.range(1, 20)
                .any(i->i %2 == 0)
                .subscribe(System.out::println);
    }

    /**
     *  序列中所有元素是否都满足 all 中的断言
     */
    public static void test_29(){

        Flux.just("abc", "ela", "ade", "pqa", "kang")
        .all(a -> a.contains("a"))
        .subscribe(System.out::println);

    }

    /**
     *  按顺序合并 跟merge有区别
     */
    public static void test_30(){

        Flux.concat(
                Flux.range(1, 3),
                Flux.range(4, 2),
                Flux.range(6, 5)
                ).subscribe(System.out::println);
    }

    /**
     *  计算 累加和
     */
    public static void test_31(){

        Flux.range(1, 10)
                .reduce(Integer::sum)
                .subscribe(System.out::println);

    }

    /**
     *  设置一个累加初始值
     */
    public static void test_32(){

        Flux.range(1, 10)
                .reduce(5, Integer::sum)
                .subscribe(System.out::println);
    }

    /**
     * onSubscribe
     * onNext
     * 1
     * onNext
     * 2
     * onNext
     * 3
     * onNext
     * 4
     * onComplete
     */
    public static void test_33(){

        Flux.just(1,2,3,4)
                .subscribe(new Subscriber<Integer>() {
                    Subscription sub;
                    @Override
                    public void onSubscribe(Subscription s) {
                        System.out.println("onSubscribe");;
                        sub = s;
                        // 每次消费一个元素
                        sub.request(1);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println("onNext");
                        System.out.println(integer);
                        sub.request(1);
                    }

                    @Override
                    public void onError(Throwable t) {
                        System.out.println("onError");;
                        t.printStackTrace();
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete");;
                    }
                });

    }

    /**
     *  结果同上
     */
    public static void test_34(){

        Flux.just(1,2,3,4)
                .subscribe(new BaseSubscriber<Integer>() {
                    Subscription sub;

                    @Override
                    protected void hookOnSubscribe(Subscription s) {
                        System.out.println("onSubscribe");;
                        sub = s;
                        // 每次消费一个元素
                        sub.request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        System.out.println("onNext");
                        System.out.println(value);
                        sub.request(1);
                    }

                    @Override
                    protected void hookOnComplete() {
                        System.out.println("onComplete");;
                    }
                });

    }

    public static void test_35(){

        Flux.interval(Duration.ofMillis(1000), Duration.ofMillis(1000))
                .timeout(Duration.ofMillis(1000))
                .subscribe(System.out::println);

    }

    public static void test_36(){

        System.out.println(Flux.interval(Duration.ofMillis(1000), Duration.ofMillis(1000))
                .blockFirst(Duration.ofMillis(1000)).intValue());

    }

    public static void test_37(){

        Flux.interval(Duration.ofMillis(1000), Duration.ofMillis(1000))
                .timeout(Duration.ofMillis(1000))
                .log()
                .subscribe(System.out::println);

    }

    public static void test_38(){

        Hooks.onOperatorDebug();
        Mono.just(0).map(x -> 1 / x)
                .checkpoint("debug").subscribe(System.out::println);
    }

    /**
     * [1, 2, 3, 4, 5, 6]
     * [7]
     * [8]
     * [9]
     * [10]
     */
    public static void test_39(){

        Flux.range(1, 10)
                .bufferUntil(i->i<5)
                .subscribe(System.out::println);

    }

    /**
     *  [6, 7, 8, 9, 10]
     */
    public static void test_40(){
        Flux.range(1, 10)
                .bufferWhile(i->i<5)
                .subscribe(System.out::println);
    }

    public static void test_41(){
        Flux.range(1,5)
                .map(i -> i*i)
                .subscribe(System.out::println);
    }

    public static void test_42(){

        Flux.just("hello", "world")
                .flatMap(w->Flux.just(w.split("")))
                .subscribe(System.out::print);

    }

    public static void test_43(){

        String[] words = new String[]{"hello", "world"};

        Arrays.stream(words)
                .flatMap(s -> Stream.of(s.split("")))
                .forEach(System.out::print);

    }

    public static void test_44(){
        Flux.just(1, 5)
            .flatMap(x -> Mono.just(x * x))
            .subscribe(System.out::println);

    }

    public static void main(String[] args) {

        test_21();



    }

}
